package org.common.utils.rpc.core;
import org.common.utils.rpc.entity.RpcRequest;
import org.common.utils.rpc.entity.RpcResponse;
import org.common.utils.rpc.exception.RpcRemoteInvokeException;
import org.common.utils.rpc.exception.RpcTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;

/**
 * RPC消息请求池
 * @author zhouzhibing
 * @date 2025/11/16 18:32
 */
public class RpcRequestPool {

    private static final Logger log = LoggerFactory.getLogger(RpcRequestPool.class);
    /**
     * 请求消息池 <请求ID，请求对象>
     */
    private final Map<Long , RpcRequest> requestMap = new ConcurrentHashMap<>();
    /**
     * 定时任务执行器对象
     */
    private final ScheduledExecutorService schedule;

    /**
     * 创建RPC消息请求池
     * @param schedule 定时任务执行器
     */
    public RpcRequestPool(ScheduledExecutorService schedule) {
        Objects.requireNonNull(schedule);
        int checkTime = Const.CHECK_TIME;
        this.schedule = schedule;
        schedule.scheduleWithFixedDelay(this::checkTimeout , checkTime , checkTime , java.util.concurrent.TimeUnit.MILLISECONDS);
    }

    /**
     * 添加请求
     * @param request 请求参数
     */
    public void addRequest(RpcRequest request) {
        requestMap.put(request.getRequestId() , request);
    }

    /**
     * 添加响应
     * @param response 响应对象
     */
    public void addResponse(RpcResponse response) {
        Objects.requireNonNull(response);
        long requestId = response.getRequestId();
        RpcRequest request = requestMap.remove(requestId);
        if(request == null) {
            // 响应结果不存在，则忽略，有些请求不需要响应，不向这里面添加，所以这里忽略。
//            log.warn("request not exist, requestId = {} , response = {}" , requestId , response);
            return;
        }

        // 获取future
        CompletableFuture<Object> future = request.getFuture();
        if(future == null) {
            log.warn("future not exist, requestId = {}" , requestId);
            return;
        }

        // 处理异常
        String error = response.getError();
        if(error != null) {
            future.completeExceptionally(new RpcRemoteInvokeException(request , error));
            return;
        }

        Object result = response.getResult();
        request.getFuture().complete(result);
    }

    /**
     * 检查请求超时
     */
    public void checkTimeout() {
        int size = requestMap.size();
        log.info("checkTimeout , request count : {}" , size);

        //如果列表是空的
        if(size == 0) {
            return;
        }

        long nowTime = System.currentTimeMillis();
        int timeout = Const.REQUEST_TIMEOUT;
        Iterator<RpcRequest> iterator = requestMap.values().iterator();
        while (iterator.hasNext()) {
            RpcRequest request = iterator.next();
            CompletableFuture<Object> future = request.getFuture();
            if(future == null) {
                log.info("checkTimeout , future not exist, request = {}" , request);
                continue;
            }

            long gapTime = nowTime - request.getCreateTime();
            if(gapTime >= timeout) {
                request.getFuture().completeExceptionally(new RpcTimeoutException(request));
                iterator.remove();
            }
        }
    }
}
